package org.spark.example

import com.sun.deploy.uitoolkit.impl.fx.Utils
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
  * @Author: chengjj
  * @Date: 2020-09-15
  */
object StreamingWordCount {
    def main(args: Array[String]): Unit = {
        var utils : Utils = new Utils;
        val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("WordCount")
        val ssc: StreamingContext = new StreamingContext(conf,Seconds(3))
        // 2. 从数据源创建一个流： socket，rdd队列，自定义接收器 ，kafka(重点)
        val rdds: mutable.Queue[RDD[String]] = mutable.Queue[RDD[String]]()
        val sourceStream: InputDStream[String] = ssc.queueStream(rdds, false);
        sourceStream.print();
        var result = sourceStream.flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _)

        result.saveAsTextFiles("E:/test/spark/stream")
        ssc.start();

        val sc = ssc.sparkContext
        while (true) {
            rdds.enqueue(sc.parallelize(1 to 100).map(_+""));
            Thread.sleep(10)
        }
    }
}
